package cn.xueden.service;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

@Slf4j
@Component
public class RocketMQProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;


    private static final String topic = "RLT_TEST_TOPIC";

    private static final String tag="RLT_TEST_TAG";

    /**
     * 发送同步消息（阻塞当前线程，等待broker响应发送结果，这样不太容易丢失消息）
     * （msgBody也可以是对象，sendResult为返回的发送结果）
     */
    public SendResult sendMsg(String msgBody) {
        SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build());
        log.info("【sendMsg】sendResult={}", JSON.toJSONString(sendResult));
        return sendResult;
    }



    public void sendMsgAsync(String msgBody) {
        Message message = new Message(topic, tag, msgBody.getBytes(StandardCharsets.UTF_8));
        rocketMQTemplate.asyncSend(topic, (org.springframework.messaging.Message<?>)message, new SendCallback() {
           @Override
           public void onSuccess(SendResult sendResult) {
               log.info("成功");
           }

           @Override
           public void onException(Throwable throwable) {
               log.info("失败");
           }
       },60,1);
    }

}
